package rpc.netty.tcp;


import com.google.gson.Gson;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import java.util.HashMap;
import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rpc.bootstrap.spring.SpringBootstrap;
import rpc.bootstrap.springboot.Bootstrap;
import rpc.core.FlowReq;
import rpc.netty.tcp.entity.MessageHeader;
import rpc.netty.tcp.entity.RpcMessage;

@Sharable
public class NettServeryHandler extends ChannelDuplexHandler {

  private static final Logger logger = LoggerFactory.getLogger(NettServeryHandler.class);
  private static final long HEADER_SIZE = 4;

  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (Objects.isNull(msg)) {
      return;
    }
    if (msg instanceof RpcMessage) {
      RpcMessage rpcMessage = (RpcMessage) msg;
      MessageHeader messageHeader = rpcMessage.getMessageHeader();
      long beginTime = System.nanoTime();
      long qs = beginTime - System.nanoTime();
      logger.info("threadName:{} id {} rpcMessage:{} qs:{} ots {}", Thread.currentThread().getName(), ctx.channel().id().asShortText(),
          new Gson().toJson(rpcMessage), qs, System.nanoTime() - beginTime);
      Bootstrap bootstrap = SpringBootstrap.getInstance().getRpcApp().getBootstrap();
      String flowName = rpcMessage.getServiceId() + "." + rpcMessage.getMsgId();
      bootstrap.invokeFlow(flowName, new FlowReq(rpcMessage.getData())).thenAccept(res -> {
        RpcMessage retRpcMessage = new RpcMessage(0, rpcMessage.getServiceId(),
            rpcMessage.getMsgId(), res.toMap(), new HashMap<>(), messageHeader);
        ctx.channel().writeAndFlush(retRpcMessage);
      });
    }
  }

  @Override
  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
      throws Exception {

    super.write(ctx, msg, promise);
  }

}